/*
 * Copyright 2015 MingxingWang

 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at

 *     http://www.apache.org/licenses/LICENSE-2.0

 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.audaque.graphdb

import java.io._

import com.audaque.graphdb.data.GraphSource
import com.audaque.graphdb.index.LuceneIndex
import com.audaque.graphdb.util.GraphDBUtil

import scala.collection.JavaConversions._
import scala.collection.mutable

class LGraphDB(val rootDir: File) {

  var openedGraphs: mutable.HashMap[String, LGraph] = new mutable.HashMap[String, LGraph]()
  var closedGraphs: mutable.HashSet[String] = new mutable.HashSet[String]()

  if (rootDir.exists() && rootDir.isDirectory) {
    for (file <- rootDir.listFiles()) {
      closedGraphs += file.getName
    }
  }

  /**
   * 创建一个图
   * @param graphName  图的名称
   * @param ignoreIfExist  是否需要判断图是否存在
   */
  def createGraph(graphName: String, ignoreIfExist: Boolean = true): Boolean = {
    val path = new File(rootDir, graphName)
    if (path.exists()) {
      if (!ignoreIfExist) {
        return false
      }
      //        throw new IllegalStateException("graph already exists: " + graphName)
    } else {
      try {
        if (!path.mkdirs()) {
          return false
          //        throw new IOException("fail to create path: " + path)
        }
      } catch {
        case _: Throwable => return false
      }

    }
    closedGraphs.add(graphName)
    true
  }

  /**
   * 删除图数据库中的所有图
   */
  def clearGraphDB() {
    val openGraphNames = openedGraphs.keySet
    for (graphName <- openGraphNames)
      dropGraph(graphName)
    val closeGraphNames = closedGraphs
    for (graphName <- closeGraphNames)
      dropGraph(graphName)
  }

  /**
   * 删除指定的图
   * @param graphName  图名称
   * @param ignoreIfNotExist  是否需要判断图是否存在
   */
  def dropGraph(graphName: String, ignoreIfNotExist: Boolean = true) {
    val path = new File(rootDir, graphName)
    if (!path.exists()) {
      if (!ignoreIfNotExist)
        throw new IllegalStateException(graphName + " does not exist!")
    } else {
      deleteDirectory(path)
    }
    if (closedGraphs.contains(graphName)) {
      closedGraphs.remove(graphName)
    } else if (openedGraphs.contains(graphName)) {
      openedGraphs.get(graphName).get.close()
      openedGraphs.remove(graphName)
    }
  }

  /**
   * 打开一个图，即将图加载到内存中，在对图进行各种查询操作之前都需要将图加载到内存中
   * @param graphName  图名
   * @param openIndexWay  打开索引的方式，有打开到内存和从磁盘直接打开两种方式
   * @return  返回一个可供查询操作的图
   */
  def openGraph(graphName: String, openIndexWay: String = "ROM"): LGraph = {
    if (openedGraphs.contains(graphName))
      closeGraph(graphName)
    val path = new File(rootDir, graphName)
    if (!path.exists())
      throw new IllegalStateException("graph does not exist: " + graphName)

    val graph: LGraph = new LGraph(graphName, path.getAbsolutePath)
    graph.load()

    openedGraphs.put(graphName, graph)
    closedGraphs.remove(graphName)
    graph
  }

  /**
   * 关闭一个图
   * @param graphName  将图从内存中取出，图关闭后释放内存，不能对图进行查询操作
   */
  def closeGraph(graphName: String) {
    openedGraphs.get(graphName) match {
      case Some(x) =>
        x.close()
        openedGraphs.remove(graphName)
        closedGraphs.add(graphName)
      case None =>
        throw new IllegalStateException(graphName + " is not opening!")
    }
  }

  /**
   * 显示图数据库中的图的信息，包括图的名称和状态(打开和关闭)，图的索引状态(打开和关闭)
   */
  def showGraph() {
    for (graph <- openedGraphs) {
      println(graph._1 + "(opened)")
    }
    for (graphName <- closedGraphs) {
      println(graphName + "(closed)")
    }
  }

  def loadDataIntoGraph(graphName: String, data: java.util.Iterator[GraphSource]): Boolean = {
    loadDataIntoGraph(graphName, data.toArray)
  }

  def loadDataIntoGraph(graphName: String, data: Iterable[GraphSource]): Boolean = {
    if (data == null || data.isEmpty) {
      return true
    }
    val path = new File(rootDir, graphName)

    val graph: LGraph = new LGraph(graphName, path.getAbsolutePath)

    val validData = new mutable.HashMap[GraphSource, Schema]()
    data.foreach(gd =>
      if (gd.fields != null && gd.fields.length > 0 && new File(gd.srcFile).exists()) {
        val schema = graph.genSchema(gd.schName, gd.fields)
        validData += (gd -> schema)
      })
    if (validData.isEmpty) {
      return true
    }

    if (!path.exists()) {
      path.mkdirs()
    }
    var luceneIndex: LuceneIndex = null
    //判断数据加载过程中是否创建索引
    val indexFields = getIndexFields(validData)
    if (indexFields.nonEmpty) {
      val indexPath = new File(path, GraphDBConf.IndexPath)
      if (indexPath.exists()) deleteDirectory(indexPath)
      luceneIndex = new LuceneIndex(indexPath)
      luceneIndex.init(graph.schemaFactory, indexFields)
      graph.indexer = luceneIndex
    }
    val vertexWriter: DataOutputStream =
      new DataOutputStream(new BufferedOutputStream(new FileOutputStream(new File(path, GraphDBConf.VertexFile))))
    val edgeWriter: DataOutputStream =
      new DataOutputStream(new BufferedOutputStream(new FileOutputStream(new File(path, GraphDBConf.EdgeFile))))

    //加载数据
    var maxVid: Long = 0
    validData.foreach(gd => {
      if (gd._1.bVertexData) {
        val maxVidTmp = loadVertexData(gd._2, gd._1, vertexWriter, luceneIndex)
        if (maxVid < maxVidTmp) maxVid = maxVidTmp
      } else {
        loadEdgeData(gd._2, gd._1, edgeWriter)
      }
    })
    graph.updateMaxVid(maxVid)
    if (luceneIndex != null) {
      luceneIndex.commit()
    }
    vertexWriter.flush()
    vertexWriter.close()
    edgeWriter.flush()
    edgeWriter.close()
    graph.saveMetaData()
    true
  }

  private def loadVertexData(schema: Schema,
                             srcData: GraphSource,
                             vertexWriter: DataOutputStream,
                             luceneIndex: LuceneIndex): Long = {
    val reader = new BufferedReader(new InputStreamReader(new FileInputStream(srcData.srcFile), srcData.charset))
    if (srcData.bFirstLineComment) {
      reader.readLine()
    }
    var maxId: Long = 0
    while (reader.ready()) {
      val line = reader.readLine()
      try {
        val values: Array[String] = line.trim.split(srcData.splitChar)
        val vid = values(0).toLong
        if (maxId < vid) maxId = vid
        val prop: Property = new Property(schema)
        for (i <- 1 until values.length) {
          prop.adjustValue(i - 1, values(i))
        }
        GraphDBUtil.saveVertex(Vertex(vid, prop), vertexWriter)
        if (luceneIndex != null) luceneIndex.add(vid, prop)
      } catch {
        case e: Throwable =>
          println("Load vertex data error: " + line)
          println(e)
      }

    }
    reader.close()
    maxId
  }

  private def loadEdgeData(schema: Schema,
                           srcData: GraphSource,
                           edgeWriter: DataOutputStream) {
    val reader = new BufferedReader(new InputStreamReader(new FileInputStream(srcData.srcFile), srcData.charset))
    if (srcData.bFirstLineComment) {
      reader.readLine()
    }
    while (reader.ready()) {
      val line = reader.readLine()
      val values: Array[String] = line.trim.split(srcData.splitChar)
      val srcId = values(0).toLong
      val dstId = values(1).toLong
      val prop: Property = new Property(schema)
      for (i <- 2 until values.length) {
        prop.adjustValue(i - 2, values(i))
      }
      GraphDBUtil.saveEdge(Edge(srcId, dstId, prop), edgeWriter)
    }
    reader.close()
  }

  private def getIndexFields(data: mutable.HashMap[GraphSource, Schema]): Map[Long, Set[String]] = {
    val result = new mutable.HashMap[Long, Set[String]]()
    data.foreach(gd => {
      if (gd._1.indexFields != null && gd._1.indexFields.nonEmpty) {
        val ifs = new mutable.HashSet[String]()
        gd._1.indexFields.foreach(f => {
          if (gd._2.getFieldIndex(f) >= 0) ifs += f
        })
        if (ifs.nonEmpty) result += (gd._2.id -> ifs.toSet)
      }
    })
    result.toMap
  }

  /**
   * 创建索引
   * @param graphName   创建索引的图的名称
   * @param fieldNames   创建索引的字段名称 schema name -> indexed field names
   */
  def createIndex(graphName: String, fieldNames: Map[String, Set[String]]) {
    if (fieldNames == null || fieldNames.isEmpty) return
    val path = new File(rootDir, graphName)
    val graph = openedGraphs.get(graphName) match {
      case Some(g) => g
      case None =>
        if (closedGraphs.contains(graphName)) {
          val g = new LGraph(graphName, path.getAbsolutePath)
          g.loadMetaData()
          g
        } else {
          return
        }
    }

    val fieldNamesChecked = new mutable.HashMap[Long, Set[String]]()
    fieldNames.foreach(fs => {
      graph.schemaFactory.getSchemaByName(fs._1) match {
        case Some(sch) =>
          val validFields = new mutable.HashSet[String]()
          fs._2.foreach(field => {
            val index = sch.getFieldIndex(field)
            if (index >= 0) validFields += field
          })
          if (validFields.nonEmpty) fieldNamesChecked += (sch.id -> validFields.toSet)
        case None =>
      }
    })
    createIndexIn(graph, fieldNamesChecked.toMap)
  }

  private def createIndexIn(graph: LGraph, fieldNames: Map[Long, Set[String]]) {
    if (fieldNames == null || fieldNames.isEmpty) return
    val graphPath: File = new File(graph.path)
    val vertexFile: File = new File(graphPath, GraphDBConf.VertexFile)
    if (!vertexFile.exists()) {
      println("Warning: no vertex data for create index.")
      return
    }
    val indexFile: File = new File(graphPath, GraphDBConf.IndexPath)
    if (indexFile.exists()) deleteDirectory(indexFile)
    val index: LuceneIndex = new LuceneIndex(indexFile)
    index.init(graph.schemaFactory, fieldNames)
    val in: DataInputStream = new DataInputStream(new FileInputStream(vertexFile))
    while (in.available() > 0) {
      val sid = in.readLong()
      val vid = in.readLong()
      val sch = graph.schemaFactory.getSchemaById(sid) match {
        case Some(s) => s
        case None => null
      }
      val prop = new Property(sch)
      prop.load(in)
      index.add(vid, prop)
    }
    index.commit()
    graph.indexer = index
  }

  /**
   * 删除目录下所有文件
   * @param dir target dir
   */
  def deleteDirectory(dir: File) {
    if (!dir.exists()) return
    if (dir.isDirectory) {
      for (file <- dir.listFiles()) {
        deleteDirectory(file)
      }
    }
    dir.delete()
  }
}
